package com.jml.service;

import com.alibaba.fastjson.JSONObject;
import com.jml.config.ProducerManage;
import com.jml.config.RocketMQConfig;
import com.jml.entity.MsgEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;


@RestController
@Slf4j
public class ProducerService {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private ProducerManage producerManage;

    /**
     * 普通消息投递  单向发送,不推荐
     */
    @GetMapping("/sendMsg")
    public String sendMsg() {
        MsgEntity msg = new MsgEntity("jml" + UUID.randomUUID().toString(), 111);
        rocketMQTemplate.convertAndSend(RocketMQConfig.TOPIC_NAME, msg);
        return "投递消息 => " + msg.toString() + " => 成功";
    }

    /**
     * 同步发送
     */
    @GetMapping("/sync")
    public void sync() {
        MsgEntity msg = new MsgEntity("jml" + UUID.randomUUID().toString(), 222);
        SendResult sendResult = rocketMQTemplate.syncSend(RocketMQConfig.TOPIC_NAME, msg);
        log.info("同步发送字符串{}, 发送结果{}", msg.toString(), sendResult);
    }

    /**
     * 异步发送
     */
    @GetMapping("async")
    public void async() {
        MsgEntity msg = new MsgEntity("jml" + UUID.randomUUID().toString(), 333);
        log.info(">msg:<<" + msg);
        rocketMQTemplate.asyncSend(RocketMQConfig.TOPIC_NAME, msg, new SendCallback() {
            @Override
            public void onSuccess(SendResult var1) {
                log.info("异步发送成功{}", var1);
            }
            @Override
            public void onException(Throwable var1) {
                log.info("异步发送失败{}", var1);
            }
        });
    }

    /**
     * 消息顺序一致性效果演示
     */
    @RequestMapping("/sequentialMessage")
    public void sequentialMessage() {
        /**
         * insert update delete
         */
//        SendResult result1 = rocketMQTemplate.syncSend(RocketMQConfig.TOPIC_SEQUENTIAL, "insert");
//        log.info("insert:" + result1.toString());
//        SendResult result2 = rocketMQTemplate.syncSend(RocketMQConfig.TOPIC_SEQUENTIAL, "update");
//        log.info("update:" + result2.toString());
//        SendResult result3 = rocketMQTemplate.syncSend(RocketMQConfig.TOPIC_SEQUENTIAL, "delete");
//        log.info("delete:" + result3.toString());
        //消息key
        String uuid = UUID.randomUUID().toString();
        SendResult result1 = rocketMQTemplate.syncSendOrderly(RocketMQConfig.TOPIC_SEQUENTIAL, "insert", uuid);
        log.info("insert:" + result1.toString());
        SendResult result2 = rocketMQTemplate.syncSendOrderly(RocketMQConfig.TOPIC_SEQUENTIAL, "update", uuid);
        log.info("update:" + result2.toString());
        SendResult result3 = rocketMQTemplate.syncSendOrderly(RocketMQConfig.TOPIC_SEQUENTIAL, "delete", uuid);
        log.info("delete:" + result3.toString());
    }


    /**
     * 原生APi
     * 发送消息注意幂等问题
     * http://127.0.0.1:8000/sendMessageIdeExt
     *原生api,生产者
     * @throws Exception
     */
    @GetMapping("/sendMessageIdeExt")
    public void sendMessageIdeExt() throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        MsgEntity msg = new MsgEntity("jml" + UUID.randomUUID().toString(), 1234);
        log.info(">msg:<<" + msg);
        Message message = new Message(RocketMQConfig.TOPIC_NAME_IDE, "member", JSONObject.toJSONString(msg).getBytes());
        SendResult result = producerManage.getProducer().send(message);
        log.info(">>result:<<" + result);

    }



}
